#include "svr_skeleton.h"

extern struct svc_map SVC_MAP[];
extern unsigned int SVC_MAP_SIZE;

struct worker_ctl
{/*{{{*/
    unsigned long thrd_id;
    int worker_idx;
    EP ep;
    EP_SLOT listener_slot;
    char *respond;

    MYSQLC db;
    int *svc_count;
};/*}}}*/
static struct worker_ctl *a_worker_ctl;

static int _callsvc(struct worker_ctl *worker_ctl, char *request)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    struct svc_map *map=NULL;
    char *respond=worker_ctl->respond;
    char *body;
    char *svc;
    int res;
    int svc_idx;
    int rtn;

    CLR_BUF(respond);
    body=request;
    if((svc=para_get_item(request, &body)) == NULL ||
            (res=strlen(svc)) <= 0 || res > NAME_SIZE)
    {
        errlog("%s: worker(%d) svc(%s) invalid",
                __func__, worker_idx, (svc)?svc:SPACES);
		para_put_item(respond, SOAR_ERR_SVC_INVALID);
        return(atoi(SOAR_ERR_SVC_INVALID));
    }

    for(svc_idx=0; svc_idx<SVC_MAP_SIZE; ++svc_idx)
    {
        map=&SVC_MAP[svc_idx];
        if(strcasecmp(map->name, svc) == 0) break;
    }
    if(svc_idx == SVC_MAP_SIZE)
    {
        errlog("%s: worker(%d) proc_(%s) not found",
                __func__, worker_idx, svc);
        para_put_item(respond, SOAR_ERR_SVC_NOT_FOUND);
        rtn=atoi(SOAR_ERR_SVC_NOT_FOUND);
    }
    else
    {
        rtn=map->pfunc(cnfg_info._db_.db[worker_idx], body, respond);
        ++worker_ctl->svc_count[svc_idx];
    }

    return(rtn);
}/*}}}*/

static int _read(struct worker_ctl *worker_ctl,
        struct ep_active_slot *active_slot)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    char *request=active_slot->_u._read.msg;
    int res;

    request[active_slot->_u._read.msg_size]=ZERO;

    if(active_slot->slot == worker_ctl->listener_slot)
    {/*{{{*/
        char *body;
        char *svc;
        char para_desc[100];
        PARA para;

        body=request;
        if(!(svc=para_get_item(request, &body)) ||
                (res=strlen(svc)) <= 0 || res > NAME_SIZE)
        {
            errlog("%s: svc(%s) invalid, impossible",
                    __func__, (svc)?svc:SPACES);
            return(-1);
        }
        if(strcmp(svc, "NEWCONN") == 0)
        {/*{{{*/
            char *value;
            SOCKET new_sock;
            EP_SLOT new_slot;

            sprintf(para_desc, "sock(number:1-%d)", SHORT_NUMBER_SIZE);
            para_load(body, para_desc, &para);
            para_ref_by_name(para, "sock", &value);
            new_sock=atoi(value);
            para_free(para);

            if((res=ep_slot_register(worker_ctl->ep, new_sock, IS_FALSE,
                            &new_slot)))
            {
                errlog("%s: worker(%d) ep_slot_register fail, res:%d",
                        __func__, worker_idx, res);
                sock_close(new_sock);
                return(-1);
            }
            ep_slot_set_wrap(new_slot);
            errlog("%s: worker(%d) new slot", __func__, worker_idx);
        }/*}}}*/
        else
        {
            errlog("%s: svc(%s) not found, impossible", __func__, svc);
            return(-1);
        }
    }/*}}}*/
    else
    {/*{{{*/
        char *respond=worker_ctl->respond;
        int respond_size;

        if((res=_callsvc(worker_ctl, request)))
        {
            errlog("%s: worker(%d) _call res(%d)",
                    __func__, worker_idx, res);
        }
        respond_size=strlen(respond);
        res=ep_write(worker_ctl->ep, active_slot->slot,
                respond, &respond_size);
        if(res)
        {
            errlog("%s: worker(%d) ep_write fail, res:%d",
                    __func__, worker_idx, res);
            return(-1);
        }
    }/*}}}*/

    return(0);
}/*}}}*/

static int _disconn(struct worker_ctl *worker_ctl,
                        struct ep_active_slot *active_slot,
                        char *type)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    struct _worker_info *worker;
    int client_num;

    if(active_slot->slot == worker_ctl->listener_slot)
    {
        errlog("%s: worker(%d) %s by listener, res:%d, "
                "impossible",
                __func__, worker_idx, type,
                active_slot->_u._abort.error);
        return(-1);
    }

    worker=&stat_info._worker_.worker[worker_idx];
    pthread_mutex_lock(&stat_info._worker_.lock);      // lock
    client_num=--worker->client_num;
    pthread_mutex_unlock(&stat_info._worker_.lock);    // unlock

    errlog("%s: worker(%d) client %s, total:%d "
            "revoke slot(sock:%d, idx:%d), res:%d",
            __func__, worker_idx, type, client_num,
            active_slot->_u._abort.sock, active_slot->slot_idx,
            active_slot->_u._abort.error);
    return(0);
}/*}}}*/

static void _worker(struct worker_ctl *worker_ctl)
{/*{{{*/
#define UPDATE_SHARE_INFO_INTERVAL          10
#define CHECK_OPERATE_INFO_INTERVAL         1
#define COUNT_NUM                           1000000
    int worker_idx=worker_ctl->worker_idx;
    struct ep_active_slot *active_list;
    unsigned int quit=IS_FALSE;
    time_t last_stat=0;
    time_t last_operate=0;
    time_t now;
    struct _svc_info *svc;
    int res;

    while(!quit)
    {
        if((res=ep_poll(worker_ctl->ep, &active_list)) != 0)
        {
            errlog("%s: worker(%d) revoke listener, res:%d, "
                    "impossible",
                    __func__, worker_idx, res);
            continue;
        }
        while(active_list)
        {
            switch(active_list->type)
            {
                case EP_SLOT_READ:
                    _read(worker_ctl, active_list);
                    break;
                case EP_SLOT_ABORT:
                    _disconn(worker_ctl, active_list, "abort");
                    break;
                case EP_SLOT_TIMEOUT:
                    _disconn(worker_ctl, active_list, "timeout");
                    break;
                default:
                    errlog("%s: worker(%d) poll type(%s) invalid, "
                            "impossible",
                            __func__, worker_idx, active_list->type);
                    break;
            }
            EP_ACTIVE_LIST_NEXT(active_list);
        }

        now=curr_timestamp(NULL);

        // update client & svc info in stat_info
        if(now - last_stat > UPDATE_SHARE_INFO_INTERVAL)
        {
            int worker_idx;

            pthread_mutex_lock(&stat_info._svc_.lock); //lock
            for(worker_idx=0; worker_idx<SVC_MAP_SIZE; ++worker_idx)
            {
                if(worker_ctl->svc_count[worker_idx] > 0)
                {
                    svc=&stat_info._svc_.svc[worker_idx];
                    svc->count+=worker_ctl->svc_count[worker_idx];
                    if(svc->count >= COUNT_NUM)
                    {
                        svc->loop+=svc->count/COUNT_NUM;
                        svc->count%=COUNT_NUM;
                    }
                    worker_ctl->svc_count[worker_idx]=0;
                }
            }
            pthread_mutex_unlock(&stat_info._svc_.lock); //unlock

            last_stat=now;
        }

        // check operate_info
        if(now - last_operate > CHECK_OPERATE_INFO_INTERVAL)
        {
            pthread_mutex_lock(&operate_info._worker_.lock); //lock
            if((quit=operate_info._worker_.quit))
            {
                operate_info._worker_.state[worker_idx]=THRD_SHUTDOWN;
            }
            pthread_mutex_unlock(&operate_info._worker_.lock); //unlock

            last_operate=now;
        }
    }
    errlog("%s: worker(%d) shutdown", __func__, worker_idx);
    DEEP_SLEEP;
}/*}}}*/

int worker_start(SOCKET **p_sock_pair)
{/*{{{*/
    SOCKET *sock_pair;
    int worker_idx;
    int i;
    int res;

    // stat_info
    pthread_mutex_init(&stat_info._worker_.lock, NULL);
    ALLOC_MULTI_DOERR(stat_info._worker_.worker, struct _worker_info,
            cnfg_info.start,
            fprintf(stderr,
                "%s: alloc stat_info._worker_.worker fail\n", SERVER_NAME);
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.start; ++worker_idx)
    {
        struct _worker_info *worker=
            &stat_info._worker_.worker[worker_idx];
        worker->client_num=0;
    }

    pthread_mutex_init(&stat_info._svc_.lock, NULL);
    ALLOC_MULTI_DOERR(stat_info._svc_.svc, struct _svc_info,
            SVC_MAP_SIZE,
            fprintf(stderr, "%s: alloc stat_info._svc_.svc fail\n",
                SERVER_NAME);
            return(-1));
    for(i=0; i<SVC_MAP_SIZE; ++i)
    {
        struct _svc_info *svc=&stat_info._svc_.svc[i];
        svc->name=SVC_MAP[i].name;
        svc->loop=0;
        svc->count=0;
    }

    // operate_info
    pthread_mutex_init(&operate_info._worker_.lock, NULL);
    operate_info._worker_.quit=IS_FALSE;
    ALLOC_MULTI_DOERR(operate_info._worker_.state, enum thrd_state,
            cnfg_info.start,
            fprintf(stderr,
                "%s: alloc operate_info._worker_.state fail\n",
                SERVER_NAME);
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.start; ++worker_idx)
    {
        operate_info._worker_.state[worker_idx]=THRD_RUNNING;
    }

    ALLOC_MULTI_DOERR(sock_pair, SOCKET, cnfg_info.start*2,
            fprintf(stderr, "%s: alloc sock_pair fail\n", SERVER_NAME);
            return(-1));

    /*--- thread worker ---*/
    ALLOC_MULTI_DOERR(a_worker_ctl, struct worker_ctl,
            cnfg_info.start,
            fprintf(stderr, "%s: alloc a_worker_ctl fail\n", SERVER_NAME);
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.start; ++worker_idx)
    {
        struct worker_ctl *worker_ctl=&a_worker_ctl[worker_idx];
        int i;

        // open listener-worker channel
        if(sock_local_pair(&sock_pair[worker_idx*2]))
        {
            fprintf(stderr,
                    "%s: open listener-worker socket pair fail\n",
                    SERVER_NAME);
            return(-1);
        }
        worker_ctl=&a_worker_ctl[worker_idx];
        if((res=ep_create(&worker_ctl->ep)))
        {
            fprintf(stderr,
                    "%s: worker(%d)(ep_slot_register) listener fail, "
                    "errcode:%d\n",
                     SERVER_NAME, worker_idx, res);
            return(-1);
        }
        if((res=ep_slot_register(worker_ctl->ep,
                        sock_pair[worker_idx*2+1], IS_FALSE,
                        &worker_ctl->listener_slot)))
        {
            fprintf(stderr,
                    "%s: worker(%d)(ep_slot_register) listener fail, "
                    "errcode:%d\n",
                    SERVER_NAME, worker_idx, res);
            return(-1);
        }
        ep_slot_set_wrap(worker_ctl->listener_slot);

        worker_ctl->worker_idx=worker_idx;
        ALLOC_MULTI_DOERR(worker_ctl->respond, char, SOCK_MSG_SIZE,
                fprintf(stderr, "%s: alloc worker(%d) respond fail\n",
                    SERVER_NAME, worker_idx);
                return(-1));
        ALLOC_MULTI_DOERR(worker_ctl->svc_count, int, SVC_MAP_SIZE,
                fprintf(stderr, "%s: alloc worker(%d) svc_count fail\n",
                    SERVER_NAME, worker_idx);
                return(-1));
        for(i=0; i<SVC_MAP_SIZE; ++i)
            worker_ctl->svc_count[i]=0;

        if(thrd_create((THRD_FUNC_T)_worker, (void *)worker_ctl,
                    THRD_DETACH, &worker_ctl->thrd_id))
        {
            fprintf(stderr, "%s: thread worker(%d) create fail\n",
                    SERVER_NAME, worker_idx);
            return(-1);
        }
    }

    *p_sock_pair=sock_pair;
    return(0);
}/*}}}*/
